package com.codejiwei.flink.transform;

import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;

/**
 * @author jiwei
 * @description 默认生命周期方法open、close 在每个并行度上都只会被调用一次
 * @date 2023/5/19 16:37
 */
public class RichMapMultiply extends RichMapFunction<Integer, Integer> {
    @Override
    public void open(Configuration parameters) throws Exception {
//        System.out.println("RichMapMultiply open...");
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
//        System.out.println("RichMapMultiply close...");
        super.close();
    }

    @Override
    public Integer map(Integer integer) throws Exception {
//        System.out.println("RichMapMultiply map...");
        getRuntimeContext();
        return integer * integer;
    }

    @Override
    public RuntimeContext getRuntimeContext() {
        RuntimeContext runtimeContext = super.getRuntimeContext();
        int attemptNumber = runtimeContext.getAttemptNumber();
        String taskName = runtimeContext.getTaskName();
        String taskNameWithSubtasks = runtimeContext.getTaskNameWithSubtasks();
        JobID jobId = runtimeContext.getJobId();

//        System.out.println("attemptNumber: " + attemptNumber);
        System.out.println("taskName: " + taskName);
//        System.out.println("taskNameWithSubtasks: " + taskNameWithSubtasks);
        System.out.println("jobId: " + jobId);
        return runtimeContext;
    }
}
